package com.gagakuai.bpm.mqlistener.camunda.baseproxy;


import com.gagakuai.bpm.common.constants.MsgLogConstants;
import com.gagakuai.bpm.domian.entity.camuda.BrokerMessageLog;
import com.gagakuai.bpm.rpcservice.openfeign.MsgLogApi;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.cglib.proxy.Enhancer;
import org.springframework.cglib.proxy.MethodInterceptor;
import org.springframework.cglib.proxy.MethodProxy;

import java.lang.reflect.Method;
import java.util.Map;

/*
 * @program: mss_server
 * @description: 消费者代理类
 * @author: houhong
 * @create: 2022-03-25 11:19
 */
@Slf4j
public class ConsumerCglib implements MethodInterceptor {


    public Object target;

    private MsgLogApi msgLogApi;


    public Object getInstance(Object target) {

        this.target = target;
        Enhancer enhancer = new Enhancer();
        //代理类
        enhancer.setSuperclass(this.target.getClass());
        //设置回调方法
        enhancer.setCallback(this);
        //创建代理对象
        Object proxy = enhancer.create();
        return proxy;
    }

    @Override
    public Object intercept(Object object, Method method, Object[] args, MethodProxy proxy) throws Throwable {

        Object result = null;

        Message message = (Message) args[0];
        Channel channel = (Channel) args[1];

        String correlationId = getCorrelationId(message);

        if (isConsumed(correlationId)) {
            log.info("重复消费, correlationId: {}", correlationId);
            return null;
        }

        MessageProperties properties = message.getMessageProperties();
        long tag = properties.getDeliveryTag();
        try {
            //实现真正的业务逻辑
            result = proxy.invokeSuper(object, args);
            //确认消息已经被消费
            msgLogApi.changeBrokerMessageLogStatus(correlationId, MsgLogConstants.BPM_WITH_DEAW_CONSUME_SUCCESS);
            // 消费确认
            channel.basicAck(tag, false);
        }catch (Exception e){

            log.error("获取代理失败", e);
            // 消息重新入队
            channel.basicNack(tag, false, true);
        }finally {
            //todo 做点什么呢
        }
        return result;
    }

    /*
     * 获取CorrelationId
     *
     * @param message
     * @return
     */
    private String getCorrelationId(Message message) {

        String correlationId = null;

        MessageProperties properties = message.getMessageProperties();
        Map<String, Object> headers = properties.getHeaders();
        for (Map.Entry entry : headers.entrySet()) {
            String key = (String) entry.getKey();
            String value = (String) entry.getValue();
            if (key.equals("spring_returned_message_correlation")) {
                correlationId = value;
            }
        }

        return correlationId;
    }

    /*
     * 消息是否已被消费
     *
     * @param correlationId
     * @return
     */
    private boolean isConsumed(String correlationId) {

        BrokerMessageLog msgLog = msgLogApi.getMessageLogById(correlationId);
        if (null == msgLog || msgLog.getStatus().equals(MsgLogConstants.BPM_WITH_DEAW_SEND_SUCCESS)) {
            return true;
        }
        return false;
    }
}